package day07

import day07.caseClass.UserVisitAction
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 6.4	需求3：页面单跳转化率统计
 *
 * 6.4.1	需求分析
 *
 * 1)	需求描述
 * 计算页面单跳转化率，什么是页面单跳转换率，比如一个用户在一次 Session 过程中访问的页面路径 3,5,7,9,10,21，
 * 那么页面 3 跳到页面 5 叫一次单跳，7-9 也叫一次单跳，那么单跳转化率就是要统计页面点击的概率
 * 比如：计算 3-5 的单跳转化率，先获取符合条件的 Session 对于页面 3 的访问次数（PV）为 A，然后获取符合条件的 Session 中
 * 访问了页面 3 又紧接着访问了页面 5 的次数为 B，那么 B/A 就是 3-5 的页面单跳转化率.
 *
 * 产品经理和运营总监，可以根据这个指标，去尝试分析，整个网站，产品，各个页面的表现怎么样，是不是需要去优化产品的布局；
 * 吸引用户最终可以进入最后的支付页面。
 *
 * 数据分析师，可以此数据做更深一步的计算和分析。
 *
 * 企业管理层，可以看到整个公司的网站，各个页面的之间的跳转的表现如何，可以适当调整公司的经营战略或策略。
 *
 * 在该模块中，需要根据查询对象中设置的 Session 过滤条件，先将对应得 Session 过滤出来，然后根据查询对象中设置的页面路径，
 * 计算页面单跳转化率，比如查询的页面路径为：3、5、7、8，那么就要计算 3-5、5-7、7-8 的页面单跳转化率。
 *
 * 需要注意的一点是，页面的访问时有先后的，要做好排序。
 *
 * 2)	思路分析
 * 	读取原始数据
 * 	将原始数据映射为样例类
 * 	将原始数据根据session进行分组
 * 	将分组后的数据根据时间进行排序（升序）
 * 	将排序后的数据进行结构的转换(pageId,1)
 * 	计算分母-将相同的页面id进行聚合统计(pageId,sum)
 * 	计算分子-将页面id进行拉链，形成连续的拉链效果，转换结构(pageId-pageId2,1)
 * 	将转换结构后的数据进行聚合统计(pageId-pageId2,sum)
 * 	计算页面单跳转换率
 */
object Spark01_TopN_Req3 {

  def main(args: Array[String]): Unit = {
    // 1.创建SparkConf并设置App名称
    val conf: SparkConf = new SparkConf().setAppName("SparkCoreTest").setMaster("local[*]")

    // 2.创建SparkContext，该对象是提交Spark App的入口
    val sc: SparkContext = new SparkContext(conf)

    val actionRDD: RDD[String] = sc.textFile(path = "E:\\学习资料\\user_visit_action.txt")

    // 3.读取原始数据，将原始数据映射为样例类
    val userVisitActionRDD: RDD[UserVisitAction] = TopNUtils.getUserVisitActionRdd(actionRDD)

    // 4.计算分母
    // 将相同的页面id进行聚合统计(pageId,sum)
    val pageIdSumRdd: RDD[(Long, Long)] = userVisitActionRDD.map(userVisitAction => (userVisitAction.pageId, 1L)).reduceByKey(_ + _)

    // 5.计算分子
    // 5.1 根据sessionId进行分组
    val sessionGroupRdd: RDD[(String, Iterable[UserVisitAction])] = userVisitActionRDD.groupBy(_.sessionId)

    // 5.2 将分组后的数据根据时间进行排序（升序）
    val sortedUserVisitActionRdd: RDD[(String, List[UserVisitAction])] = sessionGroupRdd.mapValues {
      userVisitActions => userVisitActions.toList.sortWith((left, right) => left.actionTime < right.actionTime)
    }

    // 5.3 转换结构(pageId, 1)
    val sessionPageRdd: RDD[(String, List[(Long, Int)])] = sortedUserVisitActionRdd.mapValues {
      actions: List[UserVisitAction] => actions.map(action => (action.pageId, 1))
    }

    // 5.4 将页面id进行拉链，形成连续的拉链效果，转换结构(pageId1-pageId2,1)
    val sessionZipRdd: RDD[(String, List[(String, Int)])] = sessionPageRdd.mapValues {
      pageIdList: List[(Long, Int)] => {
        val pageFlows: List[((Long, Int), (Long, Int))] = pageIdList.zip(pageIdList.tail)
        pageFlows.map {
          case ((pageId1, _), (pageId2, _)) => (pageId1 + "-" + pageId2, 1)
        }
      }
    }

    // 5.5 将转换结构后的数据进行聚合统计(pageId1-pageId2,sum)
    val flatMapRdd: RDD[(String, Int)] = sessionZipRdd.map(_._2).flatMap(list => list)
    val reduceRdd: RDD[(String, Int)] = flatMapRdd.reduceByKey(_ + _)

    // 5.6 计算页面单跳转换率
    val pageIdSumBroadcast: Broadcast[Map[Long, Long]] = sc.broadcast(pageIdSumRdd.collect().toMap)
    reduceRdd.foreach {
      case (pageFlow, numerator) =>
        val pageIdSumMap: Map[Long, Long] = pageIdSumBroadcast.value
        val pageIds: Array[String] = pageFlow.split("-")
        val pageId1: Long = pageIds(0).toLong
        // 为了避免分母不存在，这里默认值给1
        val denominator: Long = pageIdSumMap.getOrElse(pageId1, 1L)
        println(pageFlow + "=" + numerator.toDouble / denominator)
    }

    // 6.关闭连接
    sc.stop()
  }
}























